
//****** this is the final one :)

package org.apache.james.mailbox.cassandra;

import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.UUID;
import javax.mail.Flags;
import javax.mail.Flags.Flag;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.TBinaryProtocol;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.james.mailbox.MailboxPath;
import org.apache.james.mailbox.cassandra.mail.CMessage;
import org.apache.james.mailbox.cassandra.mail.model.CMailbox;
import org.apache.james.mailbox.store.mail.model.Mailbox;
import org.apache.james.mailbox.store.mail.model.Message;
import org.apache.james.mailbox.store.mail.model.Property;
import org.apache.james.mailbox.store.user.model.Subscription;
//import org.apache.openjpa.lib.util.Bytes;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;


public class CUtils {
    //private static String namespace;
     static ColumnOrSuperColumn result;
     static Column cname;
    // TODO: swith to a bit wise implementation of flags.
    private static byte[] user;
    private static byte[] name;
    private static byte[] namespace;
    private static byte[] highest_mode_seq;
    private static byte[] last_uuid;
    private static byte[] message_count;
    

    /**
     * Creates a Mailbox object from a HBase Result object.
     * @param result a result of a HBase Get operation 
     * @return a Mailbox object
     */
    public static Mailbox<UUID> mailboxFromResult(String key,String keyspace/*pass somthing which contain rowkey for mailbox*/) throws TTransportException, CharacterCodingException, InvalidRequestException, TException, NotFoundException, UnavailableException, TimedOutException {
        
        /*Get mailbox namespace,user and name using rowkey */
        
         TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
         TProtocol protocol = new TBinaryProtocol(transport);
         transport.open();
         
         
         Cassandra.Client client = new Cassandra.Client(protocol);
         client.set_keyspace(keyspace);       	
         ColumnPath path = new ColumnPath(ByteBufferUtil.string(CNames.MAILBOX_CF));
       
 	 path.setColumn(CNames.MAILBOX_NAMESPACE);
	
         result = client.get(ByteBufferUtil.bytes(key),path,ConsistencyLevel.ONE);
 	 cname =  result.getColumn();
          
         if(cname != null)
          {
              namespace=cname.getValue();
          }
        
         path.setColumn(CNames.MAILBOX_USER);
	
         result = client.get(ByteBufferUtil.bytes(key),path,ConsistencyLevel.ONE);
 	 cname =  result.getColumn();
          
         if(cname != null)
          {
           user=cname.getValue();
          }
           
         path.setColumn(CNames.MAILBOX_NAME);
	
         result = client.get(ByteBufferUtil.bytes(key),path,ConsistencyLevel.ONE);
 	 cname =  result.getColumn();
          
          if(cname != null)
          {
                 
                 name=cname.getValue();
                 
          }
          
          path.setColumn(CNames.MAILBOX_HIGHEST_MODSEQ);
	
         result = client.get(ByteBufferUtil.bytes(key),path,ConsistencyLevel.ONE);
 	 cname =  result.getColumn();
          
          if(cname != null)
          {
                 
                 highest_mode_seq=cname.getValue();
                 
          }
          
          path.setColumn(CNames.MAILBOX_LASTUID);
	
         result = client.get(ByteBufferUtil.bytes(key),path,ConsistencyLevel.ONE);
 	 cname =  result.getColumn();
          
          if(cname != null)
          {
                 
                 last_uuid=cname.getValue();
                 
          }
          
        path.setColumn(CNames.MAILBOX_MESSAGE_COUNT);
	
         result = client.get(ByteBufferUtil.bytes(key),path,ConsistencyLevel.ONE);
 	 cname =  result.getColumn();
          
          if(cname != null)
          {
                 
                 message_count=cname.getValue();
                 
          }        
        
        /*
         namespace=get namespace value from mailbox cf and namespace column
         name=get name value from " 
         user=get user value -"-;
         */
          
          
        MailboxPath p = new MailboxPath(namespace.toString(),user.toString(),name.toString());
        CMailbox mailbox = new CMailbox(p,ByteBufferUtil.toLong(CNames.MAILBOX_UIDVALIDITY));
        byte b[]=ByteBufferUtil.getArray(ByteBufferUtil.bytes(key));
        UUID ui=UUIDFromRowKey(b);
        mailbox.setMailboxId(ui);
        mailbox.setHighestModSeq(ByteBufferUtil.toLong(ByteBufferUtil.bytes(highest_mode_seq.toString())));
        mailbox.setLastUid(ByteBufferUtil.toLong(ByteBufferUtil.bytes(last_uuid.toString())));
        mailbox.setMessageCount(ByteBufferUtil.toLong(ByteBufferUtil.bytes(message_count.toString())));
        return mailbox;
    }

    /**
     
     */
    
    public static byte[] mailboxRowKey(UUID uuid) {
        
        long msb = uuid.getMostSignificantBits();
        long lsb = uuid.getLeastSignificantBits();
        byte[] buffer = new byte[16];

        for (int i = 0; i < 8; i++) {
                buffer[i] = (byte) (msb >>> 8 * (7 - i));
        }
        for (int i = 8; i < 16; i++) {
                buffer[i] = (byte) (lsb >>> 8 * (7 - i));
        }

        return buffer;
    }

    /**
     * Returns a UUID from the a byte array.
     * @param rowkey
     * @return 
     */
    public static UUID UUIDFromRowKey(byte[] byteArray) {
        long msb = 0;
        long lsb = 0;
        for (int i = 0; i < 8; i++)
                msb = (msb << 8) | (byteArray[i] & 0xff);
        for (int i = 8; i < 16; i++)
                lsb = (lsb << 8) | (byteArray[i] & 0xff);
        UUID reslt = new UUID(msb, lsb);

        return reslt;
    }

    /**
     * This method will create a object of a row of mailbox column family
     * the row object will have all value of column from mailbox object passed
     * to the method
     * @return a Put object
     */
    public static Cassandra.Client toPut(CMailbox mailbox) throws TTransportException, InvalidRequestException, TException, CharacterCodingException, UnavailableException, TimedOutException {
        //create a pointing object first using mailbox.getMailboxId()
        // we don't store null values and we don't restore them. it's a column based store.
        
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();                 
        Cassandra.Client client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace);      
        Column c=new Column();        
        ColumnParent parent = new ColumnParent(ByteBufferUtil.string(CNames.MAILBOX_CF));
        byte[] key=mailboxRowKey(mailbox.getMailboxId());
        String k=key.toString();
        if (mailbox.getName() != null) {
            
            c.setName(CNames.MAILBOX_NAME);
            client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
            
            //check if name is not null 
            //add name object to MAILBOX_CF, MAILBOX_NAME, mailbox.getName()
        }

        if (mailbox.getUser() != null) {
            c.setName(CNames.MAILBOX_USER);
            byte[] gu=ByteBufferUtil.getArray(ByteBufferUtil.bytes(mailbox.getUser()));
            c.setValue(gu);
            client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
            //check if user is not null 
            //add user object to MAILBOX_CF, MAILBOX_USER, mailbox.getUser()
        }
        if (mailbox.getNamespace() != null) {
            c.setName(CNames.MAILBOX_NAMESPACE);
            c.setValue(ByteBufferUtil.bytes(mailbox.getNamespace()));
            client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
            
            //check if namespace is not null 
            //add namespace object to MAILBOX_CF, MAILBOX_NAMESPACE,mailbox.getNamespace()
        }
        
        //add lastuuid object to MAILBOX_CF, MAILBOX_LASTUID, mailbox.getLastUid()
        c.setName(CNames.MAILBOX_LASTUID);
        c.setValue(ByteBufferUtil.bytes(mailbox.getLastUid()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //add uidvalidity object to MAILBOX_CF, MAILBOX_UIDVALIDITY,mailbox.getUidValidity()
        c.setName(CNames.MAILBOX_UIDVALIDITY);
        c.setValue(ByteBufferUtil.bytes(mailbox.getUidValidity()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //add highest mod seq object to MAILBOX_CF, MAILBOX_HIGHEST_MODSEQ,mailbox.getHighestModSeq()
        c.setName(CNames.MAILBOX_HIGHEST_MODSEQ);
        c.setValue(ByteBufferUtil.bytes(mailbox.getHighestModSeq()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //add message count object to MAILBOX_CF, MAILBOX_MESSAGE_COUNT,mailbox.getMessageCount()
        c.setName(CNames.MAILBOX_MESSAGE_COUNT);
        c.setValue(ByteBufferUtil.bytes(mailbox.getMessageCount()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //return object
          return client;  
    }

    /**
     * Transforms only the metadata into a Put object. The rest of the message will
     * be transfered using multiple Puts if size requires it. 
     * @param message
     * @return a put that contains all metadata information.
     */
    public static Cassandra.Client metadataToPut(Message<UUID> message) throws TTransportException, InvalidRequestException, CharacterCodingException, TException, UnavailableException, TimedOutException {
        //change Message<UUID> message to two seprate key
        //combine it to form row key if possible else we will create seprate column for both
        //Create pointing object with messageRowKey(message)
        // we store the message uid and mailbox uid in the row key
        // store the meta data
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();                 
        Cassandra.Client client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace);      
        Column c=new Column();        
        ColumnParent parent = new ColumnParent(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
        byte[] b=messageRowKey(message.getMailboxId(),message.getUid());
        String k=b.toString();
                 
        //put.add(MESSAGES_META_CF, MESSAGE_MODSEQ, Bytes.toBytes(message.getModSeq()));
        c.setName(CNames.MESSAGE_MODSEQ);
        c.setValue(ByteBufferUtil.bytes(message.getModSeq()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //put.add(MESSAGES_META_CF, MESSAGE_INTERNALDATE, Bytes.toBytes(message.getInternalDate().getTime()));
        c.setName(CNames.MESSAGE_INTERNALDATE);
        c.setValue(ByteBufferUtil.bytes(message.getInternalDate().getTime()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //put.add(MESSAGES_META_CF, MESSAGE_MEDIA_TYPE, Bytes.toBytes(message.getMediaType()));
        c.setName(CNames.MESSAGE_MEDIA_TYPE);
        c.setValue(ByteBufferUtil.bytes(message.getMediaType()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //put.add(MESSAGES_META_CF, MESSAGE_SUB_TYPE, Bytes.toBytes(message.getSubType()));
        c.setName(CNames.MESSAGE_SUB_TYPE);
        c.setValue(ByteBufferUtil.bytes(message.getSubType()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //put.add(MESSAGES_META_CF, MESSAGE_CONTENT_OCTETS, Bytes.toBytes(message.getFullContentOctets()));
        c.setName(CNames.MESSAGE_CONTENT_OCTETS);
        c.setValue(ByteBufferUtil.bytes(message.getFullContentOctets()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        //put.add(MESSAGES_META_CF, MESSAGE_BODY_OCTETS, Bytes.toBytes(message.getBodyOctets()));
        c.setName(CNames.MESSAGE_BODY_OCTETS);
        c.setValue(ByteBufferUtil.bytes(message.getBodyOctets()));
        client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);
        
        if (message.getTextualLineCount() != null) {
            //put.add(MESSAGES_META_CF, MESSAGE_TEXT_LINE_COUNT, Bytes.toBytes(message.getTextualLineCount()));
            c.setName(CNames.MESSAGE_TEXT_LINE_COUNT);
            c.setValue(ByteBufferUtil.bytes(message.getTextualLineCount()));
            client.insert(ByteBufferUtil.bytes(k), parent,c,ConsistencyLevel.ONE);    
        
        }
 // store system flags in meta and user flags in uflags to avoid name clashes
        Flags flags = message.createFlags();
        // system flags
        if (flags.contains(Flag.ANSWERED)) {
            c.setName(FlagConvertor.FLAGS_ANSWERED);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_ANSWERED, MARKER_PRESENT);
        }
        if (flags.contains(Flag.DELETED)) {
            c.setName(FlagConvertor.FLAGS_DELETED);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_DELETED, MARKER_PRESENT);
        }
        if (flags.contains(Flag.DRAFT)) {
            c.setName(FlagConvertor.FLAGS_DRAFT);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_DRAFT, MARKER_PRESENT);
        }
        if (flags.contains(Flag.FLAGGED)) {
            c.setName(FlagConvertor.FLAGS_FLAGGED);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_FLAGGED, MARKER_PRESENT);
        }
        if (flags.contains(Flag.RECENT)) {
            c.setName(FlagConvertor.FLAGS_RECENT);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_RECENT, MARKER_PRESENT);
        }
        if (flags.contains(Flag.SEEN)) {
            c.setName(FlagConvertor.FLAGS_SEEN);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_SEEN, MARKER_PRESENT);
        }
        if (flags.contains(Flag.USER)) {
            c.setName(FlagConvertor.FLAGS_USER);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_USER, MARKER_PRESENT);
        }

        // user flags
        for (String flag : flags.getUserFlags()) {
            c.setName(ByteBufferUtil.bytes(flag));
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, userFlagToBytes(flag), MARKER_PRESENT);
        }
        int propNumber = 0;
        //add the properties
        for (Property prop : message.getProperties()) {
            c.setName(ByteBufferUtil.bytes(PropertyConvertor.getQualifier(propNumber++).toString()));
            c.setValue(ByteBufferUtil.bytes(PropertyConvertor.getValue(prop).toString()));            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, getQualifier(propNumber++), getValue(prop));
        }

        return client;
    }

    /**
     * Create a row key for a message in a mailbox. The current row key is mailboxID followed by messageID.
     * Both values are fixed length so no separator is needed. 
     * Downside: we will be storing the same message multiple times, one time for each recipient.
     * @param message
     * @return a row key 
     */
    public static byte[] messageRowKey(Message<UUID> message) {
        return messageRowKey(message.getMailboxId(), message.getUid());
    }

    /**
     * Utility method to build row keys from mailbox UUID and message uid.
     * The message uid's are stored in reverse order by substracting the uid value 
     * from Long.MAX_VALUE. 
     * @param message
     * @param uid
     * @return a row key byte array
     */
    public static byte[] messageRowKey(UUID mailboxUid, long uid) {
        /**  message uid's are stored in reverse order so we will always have the most recent messages first*/
        byte[] a1= ByteBufferUtil.getArray(ByteBufferUtil.bytes(mailboxUid.getMostSignificantBits()));
        int srcPos=a1.length;
        byte[] a2= ByteBufferUtil.getArray(ByteBufferUtil.bytes(mailboxUid.getLeastSignificantBits()));
        int dstPos=(srcPos+(a2.length));
        int mxvl=(int)(Long.MAX_VALUE-uid);
        ByteBufferUtil.arrayCopy(ByteBufferUtil.bytes(a1.toString()), srcPos, ByteBufferUtil.bytes(a2.toString()), dstPos,mxvl);
             
         return a1;
    }

    /**
     * Utility to build row keys from mailboxUID and a value. The value is added to 
     * the key without any other opperations. 
     * @param mailboxUid
     * @param value
     * @return 
     */
    public static byte[] customMessageRowKey(UUID mailboxUid, long value) {
        
        byte[] a1= ByteBufferUtil.getArray(ByteBufferUtil.bytes(mailboxUid.getMostSignificantBits()));
        int srcPos=a1.length;
        byte[] a2= ByteBufferUtil.getArray(ByteBufferUtil.bytes(mailboxUid.getLeastSignificantBits()));
        int dstPos=(srcPos+(a2.length));
        ByteBufferUtil.arrayCopy(ByteBufferUtil.bytes(a1.toString()), srcPos, ByteBufferUtil.bytes(a2.toString()), dstPos,(int)value);
             
         return a1;
        
    }

    /**
     * Creates a HBaseMessage from a Result object. This method retrieves all information 
     * except for body and header related bytes. The message content will be loaded on demand
     * through a specialised InputStream called {@link ChunkInputStream}. 
     * IMPORTANT: the method expects a single version of each cell. Use setMaxVersions(1).
     * @param conf configuration object for HBase cluster
     * @param result the result object containing message data
     * @return a HBaseMessage instance with message metadata.
     */
    public static Message<UUID> messageMetaFromResult(String key,String keyspace) {
        CMessage message = null;
        
        //This function retrive all values of column from message meta column family by a 
        ///row id and assign to message object.
        /*
        UUID uuid = UUIDFromRowKey(Bytes.toBytes(key));
        uuid = Long.MAX_VALUE - Bytes.toLong(result.getRow(), 16);
        PropertyBuilder props = new PropertyBuilder(propList);
        props.setMediaType(mediaType);
        props.setSubType(subType);
        message = new HBaseMessage(conf, uuid, internalDate, flags, contentOctets, (int) (contentOctets - bodyOctets), props);
        message.setUid(uid);
        message.setModSeq(modSeq);
        message.setTextualLineCount(textualLineCount);
        
         */
        
        return message;
    }

    /**
     * Creates a Put object from this subscription object
     * @return Put object suitable for HBase persistence
     */
    public static Cassandra.Client toPut(Subscription subscription) throws TTransportException, InvalidRequestException, TException, CharacterCodingException, UnavailableException, TimedOutException {
        
        
        //Put put = new Put(Bytes.toBytes(subscription.getUser()));
        
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();                 
        Cassandra.Client client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace);      
        Column c=new Column();        
        ColumnParent parent = new ColumnParent(ByteBufferUtil.string(CNames.MAILBOX_CF));
        String k=subscription.getUser();
        c.setName(ByteBufferUtil.bytes(subscription.getMailbox()));
        c.setValue(CNames.MARKER_PRESENT);
        client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
        //put.add(SUBSCRIPTION_CF, Bytes.toBytes(subscription.getMailbox()), MARKER_PRESENT);
        return client;
    }

    /**
     * Utility method to transform message flags into a put opperation.
     * @param message
     * @param flags
     * @return a put object with 
     */
    public static Cassandra.Client flagsToPut(Message<UUID> message, Flags flags) throws TTransportException, InvalidRequestException, TException, CharacterCodingException, UnavailableException, TimedOutException {
        
        
        //Put put = new Put(messageRowKey(message));
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();                 
        Cassandra.Client client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace);      
        Column c=new Column();        
        ColumnParent parent = new ColumnParent(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
        byte[] b=messageRowKey(message.getMailboxId(),message.getUid());
        String k=b.toString();
        
        //system flags
        
        
        if (flags.contains(Flag.ANSWERED)) {
            c.setName(FlagConvertor.FLAGS_ANSWERED);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_ANSWERED, MARKER_PRESENT);
        } else {
            c.setName(FlagConvertor.FLAGS_ANSWERED);
            c.setValue(CNames.MARKER_MISSING);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_ANSWERED, MARKER_MISSING);
        }
        if (flags.contains(Flag.DELETED)) {
            c.setName(FlagConvertor.FLAGS_DELETED);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_DELETED, MARKER_PRESENT);
        } else {
            c.setName(FlagConvertor.FLAGS_DELETED);
            c.setValue(CNames.MARKER_MISSING);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_DELETED, MARKER_MISSING);
        }
        if (flags.contains(Flag.DRAFT)) {
            
            c.setName(FlagConvertor.FLAGS_DRAFT);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_DRAFT, MARKER_PRESENT);
        } else {
            
            c.setName(FlagConvertor.FLAGS_DRAFT);
            c.setValue(CNames.MARKER_MISSING);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_DRAFT, MARKER_MISSING);
        }
        if (flags.contains(Flag.FLAGGED)) {
            
            c.setName(FlagConvertor.FLAGS_FLAGGED);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_FLAGGED, MARKER_PRESENT);
        } else {
            c.setName(FlagConvertor.FLAGS_FLAGGED);
            c.setValue(CNames.MARKER_MISSING);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_FLAGGED, MARKER_MISSING);
        }
        if (flags.contains(Flag.RECENT)) {
            c.setName(FlagConvertor.FLAGS_RECENT);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_RECENT, MARKER_PRESENT);
        } else {
            c.setName(FlagConvertor.FLAGS_RECENT);
            c.setValue(CNames.MARKER_MISSING);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_RECENT, MARKER_MISSING);
        }
        if (flags.contains(Flag.SEEN)) {
            c.setName(FlagConvertor.FLAGS_SEEN);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_SEEN, MARKER_PRESENT);
        } else {
            c.setName(FlagConvertor.FLAGS_SEEN);
            c.setValue(CNames.MARKER_MISSING);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_SEEN, MARKER_MISSING);
        }
        if (flags.contains(Flag.USER)) {
            c.setName(FlagConvertor.FLAGS_USER);
            c.setValue(CNames.MARKER_PRESENT);            
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_USER, MARKER_PRESENT);
        } else {
            
            c.setName(FlagConvertor.FLAGS_USER);
            c.setValue(CNames.MARKER_MISSING);
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, FLAGS_USER, MARKER_MISSING);
        }
        /**TODO: user flags are not deleted this way: store them all in a single column  
         * and replace that column full.
         */
        // user flags
        for (String flag : flags.getUserFlags()) {
 
            c.setName(ByteBufferUtil.bytes(flag.toString()));
            c.setValue(CNames.MARKER_MISSING);           
            client.insert(ByteBufferUtil.bytes(k),parent,c,ConsistencyLevel.ONE);
            //put.add(MESSAGES_META_CF, userFlagToBytes(flag), MARKER_PRESENT);
        }
        return client;
    }

    public static void flagsToDelete(Message<UUID> message, Flags flags) {
        /*Delete delete = new Delete(messageRowKey(message));
        //we mark for delete flags that are not present (they will be Put'ed)
        if (flags.contains(Flag.ANSWERED)) {
            delete.deleteColumn(MESSAGES_META_CF, FLAGS_ANSWERED);
        }
        if (flags.contains(Flag.DELETED)) {
            delete.deleteColumn(MESSAGES_META_CF, FLAGS_DELETED);
        }
        if (flags.contains(Flag.DRAFT)) {
            delete.deleteColumn(MESSAGES_META_CF, FLAGS_DRAFT);
        }
        if (flags.contains(Flag.FLAGGED)) {
            delete.deleteColumn(MESSAGES_META_CF, FLAGS_FLAGGED);
        }
        if (flags.contains(Flag.RECENT)) {
            delete.deleteColumn(MESSAGES_META_CF, FLAGS_RECENT);
        }
        if (flags.contains(Flag.SEEN)) {
            delete.deleteColumn(MESSAGES_META_CF, FLAGS_SEEN);
        }
        if (flags.contains(Flag.USER)) {
            delete.deleteColumn(MESSAGES_META_CF, FLAGS_USER);
        }

        // we delete all user flags that where not in the new configuration
        for (String flag : flags.getUserFlags()) {
            delete.deleteColumn(MESSAGES_META_CF, userFlagToBytes(flag));
        }
        return delete;
    
         * 
         * /
         * 
         */
            }

    /**
     * Returns a String composed of all flags in the  parameter.
     * @param flags
     * @return a string representation of all flags
     */
    public static String flagsToString(Flags flags) {
        StringBuilder b = new StringBuilder();

        if (flags.contains(Flag.ANSWERED)) {
            b.append("ANSWERED ");
        }
        if (flags.contains(Flag.DELETED)) {
            b.append("DELETED ");
        }
        if (flags.contains(Flag.DRAFT)) {
            b.append("DRAFT ");
        }
        if (flags.contains(Flag.FLAGGED)) {
            b.append("FLAGGED ");
        }
        if (flags.contains(Flag.RECENT)) {
            b.append("RECENT ");
        }
        if (flags.contains(Flag.SEEN)) {
            b.append("SEEN ");
        }
        if (flags.contains(Flag.USER)) {
            b.append("USER ");
        }
        for (String flag : flags.getUserFlags()) {
            b.append(flag);
            b.append(" ");
        }
        return b.toString();
    }
}
